/*
 * ObjectMessageProtocolHandler.java    0.0.1    26/set/2009
 * 
 * Copyright (c) 2009 mentalsmash.org
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 * The use of the Apache License does not indicate that this project is
 * affiliated with the Apache Software Foundation.
 */ 
package org.mentalsmash.tazio.net.cs.server;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;

import org.mentalsmash.tazio.net.cs.MessageInputStream;
import org.mentalsmash.tazio.net.cs.MessageOutputStream;
import org.mentalsmash.tazio.net.cs.RuntimeCSSubsystemException;
import org.mentalsmash.tazio.net.cs.protocol.MessageHeader;
import org.mentalsmash.tazio.net.cs.protocol.NIOObjectContainer;
import org.mentalsmash.tazio.net.cs.server.impl.ChannelFacade;
import org.mentalsmash.tazio.net.cs.server.impl.InputQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * The <tt>ObjectMessageProtocolHandler</tt> class provides an <code>abstract</code> implementation
 * of {@link ProtocolHandler}, providing the basic abstractions to receive and send objects that implement
 * the {@link Serializable} interface over a channel, instead of being force to cope with <code>byte[]</code>
 * or <code>Buffer</code>s.
 * <p>
 * Concrete classes that want to subclass <tt>ObjectMessageProtocolHandler</tt> have to implement the
 * {@link #handleMessage(Serializable,MessageOutputStream)} method ( along with the 
 * {@link #registered(ChannelFacade)} and {@link #unregistered(ChannelFacade)} from the 
 * {@link ProtocolHandler} interface). 
 * </p>
 * <p>
 * {@link #handleMessage(Serializable,MessageOutputStream)} will be automatically called each time an object
 * is received from the channel. That object is passed as argument of the method, together with a reference
 * to an {@link MessageOutputStream} instance, that provides convenience methods to write back the responce
 * to the channel that originated the request.
 * </p>
 * @version    0.0.1    26/set/2009    
 * @author     Andrea Reale <andrea.reale@gmail.com>
 * @see        ProtocolHandler
 * @see        ObjectChannelWriter
 */
public abstract class ObjectMessageProtocolHandler implements ProtocolHandler {

	private final static Logger log = LoggerFactory.getLogger(ObjectMessageProtocolHandler.class);
	
	
	private MessageHeader nextHeader;
	private ByteBuffer headerBuffer;
	
	private boolean headerReceived = false;
	
	
	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.cs.server.ProtocolHandler#handleMessage(java.nio.ByteBuffer, org.mentalsmash.tazio.net.cs.server.impl.ChannelFacade)
	 */
	@Override
	public final void handleMessage(ByteBuffer message, ChannelFacade source) {
		byte[] messageArray = bufferToArray(message);
		ByteArrayOutputStream backend = null;
		MessageOutputStream messageOutput = null;
		
		MessageInputStream inputStream = new MessageInputStream(new ByteArrayInputStream(messageArray));
		Serializable payload = null;
		
		
		
		
		try {
			payload = inputStream.readMessage();
			
		} catch (ClassCastException e) {
			String errorMsg = "Received message is not an instance of " + NIOObjectContainer.class.getSimpleName();
			log.error(errorMsg,e);
			throw new RuntimeCSSubsystemException(errorMsg, e);
		} catch (ClassNotFoundException e) {
			log.error("Cannot deserialise object", e);
			throw new RuntimeCSSubsystemException(e);
		} catch (IOException e) {
			log.error("Error while retrieving object message. ", e);
		} 
		//invokes the higher level handler, passing it the payload and an instance of ObjectChannelWriter
		//to allow applications to write high-level Serializable objects to the channel that originated
		//the request
		backend = new ByteArrayOutputStream();
		messageOutput = new MessageOutputStream(backend);
		
		handleMessage(payload, messageOutput);
		
		try {
			messageOutput.flush();
			//wrapping into a ByteBuffer the contents written by the MessageOutputStramInstane
			//and enqueuing them into the output queue of the source
			source.getOutputQueue().enqueue(ByteBuffer.wrap(backend.toByteArray()));
		} catch (IOException e){
			log.error("FALTA: Really unexpected excpetion!");
			throw new RuntimeCSSubsystemException("This exception should never be thrown", e);
		} finally {
			try {
				messageOutput.close();
			} catch (IOException e) {
				//never mind
			}
		}
		
		

	}

	/**
	 * This method is invoked by the <i>CS Communication Subsystem</i> each time a Serializable object
	 * is received from the channel which the instance of this Handler is associated to.
	 * Along with the received message an instance of {@link MessageOutputStream} is provided:
	 * use its method {@link MessageOutputStream#writeMessage(Serializable)} to write back an object response to the sender.
	 * <p>
	 * This method is automatically invoked by the working thread that is associated with an incoming request.
	 * </p>
	 * @param msgReceived the object that was received from the channel associated with this handler's instance
	 * @param channelWriter an outputStream associated to the channel which sent the object 
	 */
	public abstract void handleMessage(Serializable msgReceived, MessageOutputStream channelWriter);
	
	


	/* (non-Javadoc)
	 * @see org.mentalsmash.tazio.net.cs.server.ProtocolHandler#nextMessage(org.mentalsmash.tazio.net.cs.server.impl.ChannelFacade)
	 */
	//TODO: Make this method use MessageInputStream for clearenss
	@Override
	public final ByteBuffer nextMessage(ChannelFacade source) {
		InputQueue input = source.getInputQueue();
		ByteBuffer completeMessage;
		ByteBuffer payloaBuffer;
		int payloadLength;
		
		try {
		    if (! retrieveHeader(input) ) { //tries to retrieve the header of the next message, if it can't returns null
			    return null;
		    }
		} catch (IOException e) {
		    throw new RuntimeCSSubsystemException("FATAL error while retrieving a message hedaer", e);
		}

		payloadLength = this.nextHeader.getPayloadLength();
		
		//When retrieveHeader() has returned true it will have dequeued the header from the input,
		//so there's only the payload to retrieve
		
		if(input.size() < payloadLength){
			return null;
		} 
		
		log.debug("Retrieving msg payload from the channel");
		
		completeMessage = ByteBuffer.allocate(MessageHeader.HEADER_SIZE+payloadLength); // allocating size for the complete
								                                // message
		
		payloaBuffer = input.dequeueBytes(payloadLength);
		
		
		
		completeMessage.put(headerBuffer).put(payloaBuffer);
		
		/*
		 * resetting handler's status
		 */
		this.nextHeader = null;
		headerReceived = false;
		headerBuffer = null;
		
		return completeMessage;
		
		
		
	}

	/**
	 * Reads from the input source and tries to retrieve a message header.
	 * If it manages to find it, dequeues it from the input queue, and changes the
	 * status of this protocol handler, putting it in the "Waiting for the payload"
	 * state.
	 * @param input The input queue where to search for the header
	 * @return true if the header was previously or just now received. Otherwise false
	 * @throws IOException If it fails to retrieve a valid header.
	 */
	private boolean retrieveHeader(InputQueue input) throws IOException {
		byte[] header;
		
		if (headerReceived)
			return true;
		
		if(input.size() <= MessageHeader.HEADER_SIZE) {
			return false;
		}
		
		
		headerBuffer = input.dequeueBytes(MessageHeader.HEADER_SIZE); //getting the header
		headerReceived = true;
		
		header =bufferToArray(headerBuffer);
		
		this.nextHeader = MessageHeader.readHeader(header);
		
		log.debug("Header for next message completely retrieved.");
		
		return headerReceived;		
		
	}
	
	/**
	 * Returns the entire content of a byte buffer as an array.
	 * <p>
	 * If the buffer is backed by an array returns this array: notice that in such a case
	 * modifying the array modifies the buffer too. If the buffer is not backed by
	 * an array copy the whole buffer content (from the current position to to the current limit)
	 * into a newly created array, and then restors the current position
	 * </p>
	 * @param b the buffer to get the array from
	 * @return Returns the entire content of a byte buffer as an array.
	 */
	private byte[] bufferToArray(ByteBuffer b){
		
		if(b.hasArray()) {
			return b.array();
		} else {
			b.mark();
			byte[] array = new byte[b.remaining()];
			for(int i=0;b.hasRemaining();i++) {
				array[i] = b.get();
			}

			b.reset(); //reverting the array in read-mode
			return array;
			
		}
		
		
	}
	


}
